package com.nit.netty.bean.aware;

import com.alibaba.fastjson.JSONArray;
import com.nit.netty.codec.JSONDecoder;
import com.nit.netty.codec.JSONEncoder;
import com.nit.netty.handler.NettyClientHandler;
import com.nit.netty.listener.ConnectionListener;
import com.nit.netty.manager.NettyChannelManager;
import com.nit.netty.model.Request;
import com.nit.netty.model.Response;
import com.nit.netty.sender.RequestSender;
import com.nit.netty.zookeeper.client.ZookeeperClient;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;

import javax.annotation.PreDestroy;
import java.net.SocketAddress;
import java.util.concurrent.SynchronousQueue;

/**
 * @Author: whb
 * @Description:
 * @CreateTime: 2020/8/21 13:46
 */
public class NettyClientApplicationContextAware extends ChannelInitializer<SocketChannel>
        implements RequestSender, ApplicationContextAware, InitializingBean {

    private Bootstrap bootstrap;
    private EventLoopGroup group;
    private NettyChannelManager manager;
    private NettyClientHandler handler;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.bootstrap = new Bootstrap();
        // 这里的group中的eventLoop只负责监听channel状态
        // 如果channel空闲则可以发送数据
        // 如果channel中有回传则接收数据
        this.group = new NioEventLoopGroup(1);
        this.bootstrap.group(group).
                channel(NioSocketChannel.class).
                option(ChannelOption.TCP_NODELAY, true).
                option(ChannelOption.SO_KEEPALIVE, true).
                handler(this);
        String zkAddress = applicationContext.getEnvironment().getProperty("zookeeper.address");
        ZookeeperClient zkClient = new ZookeeperClient(zkAddress);
        this.manager = new NettyChannelManager(this, zkClient);
        this.handler = new NettyClientHandler();
    }

    @Override
    public void afterPropertiesSet(){
        // TODO do something...
        manager.refresh();
    }

    @Override
    public Object send(Request request) throws InterruptedException {
        Channel channel = null;
        Class<?> clazz = null;
        try {
            clazz = Class.forName(request.getClassName());
            // 分布式下, 一个服务会部署到多台服务器, 所以这里用take轮询获得channel
            channel = manager.take(clazz.getSimpleName());
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }
        if (channel != null && channel.isActive()) {
            SynchronousQueue<Object> queue = this.handler.sendRequest(request, channel);
            // 等待结果返回, queue为空的时候, 这里会阻塞
            Object result = queue.take();
            return JSONArray.toJSONString(result);
        } else {
            Response res = new Response();
            res.setCode(502);
            assert clazz != null;
            res.setError("未正确连接到服务器.请检查相关配置信息! 服务名:" + clazz.getSimpleName());
            return JSONArray.toJSONString(res);
        }
    }

    @Override
    protected void initChannel(SocketChannel channel) throws Exception {
        ChannelPipeline pipeline = channel.pipeline();
        // 设置心跳时间 1分钟发送一次心跳
        pipeline.addLast(new IdleStateHandler(0, 0, 60));
        // 设置编码解码器
        pipeline.addLast(new JSONEncoder());
        pipeline.addLast(new JSONDecoder());
        // 设置处理器
        pipeline.addLast(this.handler);
    }

    @Override
    public Channel connect(SocketAddress address) throws InterruptedException {
        ChannelFuture future = bootstrap.connect(address);
        future.addListener(new ConnectionListener(this.manager));
        // 如果通道已经关闭, 这里会抛出异常
        Channel channel = null;
        try {
            channel = future.sync().channel();
        } catch(Exception e) {
            e.printStackTrace();
            return null;
        }
        return channel;
    }

    @PreDestroy
    public void destroy() {
        this.group.shutdownGracefully();
    }
}

